/**
 * Covalent Server Code File
 * 
 * File:        thread_helper.c
 * Brief:       BRIEF
 * Author:      Chen Kangbing
 * Email:       ckblau@outlook.com
 * Modified:    2021/10/08 17:07:32
 *
 */


#include "thread_helper.h"
#include "socket_helper.h"

#include <errno.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <pthread.h>
#include <semaphore.h>


int thread_start(pthread_t *thread_id, THREAD_BODY *thread_workbody,
                 void *thread_arg) {
    int rv = -1;
    pthread_attr_t thread_attr;

    // printf("Starting thread...\n");

    if (pthread_attr_init(&thread_attr)) {
        printf("pthread_attr_init() failed: %s\n", strerror(errno));
        goto CleanUp;
    }

    if (pthread_attr_setstacksize(&thread_attr, 120 * 1024)) {
        printf("pthread_attr_setstacksize() failed: %s\n", strerror(errno));
        goto CleanUp;
    }

    if (pthread_attr_setdetachstate(&thread_attr, PTHREAD_CREATE_DETACHED)) {
        printf("pthread_attr_setdetachstate() failed: %s\n", strerror(errno));
        goto CleanUp;
    }

    if (pthread_create(thread_id, &thread_attr, thread_workbody, thread_arg)) {
        printf("Failed to create thread: %s\n", strerror(errno));
        goto CleanUp;
    }

    rv = 0;
CleanUp:
    pthread_attr_destroy(&thread_attr);
    return rv;
}

int thread_check(WorkerThreadTypeDef *pworker) {
    if (!pworker) return 0;
    if (!pworker->thread_tid) return 0;
    int rv = pthread_kill(pworker->thread_tid, 0);
    if (rv == ESRCH || rv == EINVAL) {
        return 0;
    }
    else {
        return 1;
    }
}

int thread_init_pool(ServerTypeDef *psrv, THREAD_BODY *thread_workbody) {
    if (!psrv) return -1;

    psrv->task_queue = calloc(psrv->max_connection, sizeof(WorkerTaskTypeDef));
    if (!psrv->task_queue) return -2;
    psrv->queue_head = 0;
    psrv->queue_tail = 0;
    pthread_mutex_init(&(psrv->queue_mut), NULL);
    sem_init(&(psrv->queue_sem), 0, 0);
    sem_init(&(psrv->connection_sem), 0, psrv->max_connection);

    psrv->thread_pool = calloc(psrv->max_pool_size, sizeof(WorkerThreadTypeDef*));
    if (!psrv->thread_pool) return -3;
    for (int i = 0; i < psrv->max_pool_size; ++i) {
        psrv->thread_pool[i] = malloc(sizeof(WorkerThreadTypeDef));
        if (!psrv->thread_pool[i]) return -4;
        WorkerThreadTypeDef *pworker = psrv->thread_pool[i];
        pworker->server = psrv;
        thread_start(&(pworker->thread_tid), thread_workbody, (void *)pworker);
    }

    return 0;
}

void task_add(ServerTypeDef *psrv, WorkerTaskTypeDef *ptask) {
    sem_wait(&(psrv->connection_sem));
    pthread_mutex_lock(&(psrv->queue_mut));
    psrv->task_queue[psrv->queue_tail] = *ptask;
    psrv->queue_tail = (psrv->queue_tail + 1) % psrv->max_connection;

    int c = 0, q = 0;
    sem_getvalue(&(psrv->connection_sem), &c);
    sem_getvalue(&(psrv->queue_sem), &q);
    printf("task_add, conn = %d, queue = %d, [%d, %d]\n", c, q, psrv->queue_head, psrv->queue_tail);

    pthread_mutex_unlock(&(psrv->queue_mut));
    sem_post(&(psrv->queue_sem));
}

void task_get(WorkerThreadTypeDef *pworker) {
    ServerTypeDef *psrv = pworker->server;
    sem_wait(&(psrv->queue_sem));
    pthread_mutex_lock(&(psrv->queue_mut));
    pworker->task = psrv->task_queue[psrv->queue_head];
    memset(&(psrv->task_queue[psrv->queue_head]), 0, sizeof(WorkerTaskTypeDef));
    psrv->queue_head = (psrv->queue_head + 1) % psrv->max_connection;

    int c = 0, q = 0;
    sem_getvalue(&(psrv->connection_sem), &c);
    sem_getvalue(&(psrv->queue_sem), &q);
    printf("task_get, conn = %d, queue = %d, [%d, %d]\n", c, q, psrv->queue_head, psrv->queue_tail);

    pthread_mutex_unlock(&(psrv->queue_mut));
}

void task_finish(WorkerThreadTypeDef *pworker) {
    ServerTypeDef *psrv = pworker->server;
    sem_post(&(psrv->connection_sem));

    int c = 0, q = 0;
    sem_getvalue(&(psrv->connection_sem), &c);
    sem_getvalue(&(psrv->queue_sem), &q);
    printf("task_finish, conn = %d, queue = %d, [%d, %d]\n", c, q, psrv->queue_head, psrv->queue_tail);
}

void thread_destroy(WorkerThreadTypeDef *pworker) {
    if (!pworker) return;
    if (pworker->thread_tid) pthread_kill(pworker->thread_tid, SIGKILL);
    task_shutdown(&(pworker->task));
    worker_reset(pworker);
    free(pworker);
    pworker = NULL;
}

